package cn.kennylee.codehub.mq.lip.integration.base;

import cn.hutool.core.util.StrUtil;
import cn.hutool.json.JSON;
import cn.hutool.json.JSONArray;
import cn.hutool.json.JSONUtil;
import lombok.AccessLevel;
import lombok.Getter;
import lombok.Setter;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.spring.annotation.RocketMQMessageListener;
import org.slf4j.MDC;
import org.springframework.lang.NonNull;

import java.util.List;
import java.util.Objects;

/**
 * <p> 抽象消息监听器 </p>
 * <p>Created on 2025/4/16.</p>
 *
 * @param <T> 消息类型
 * @author kennylee
 * @since 0.0.1
 */
@Slf4j
public abstract class AbstractMqListener<T> {

    /**
     * 最大消息重试次数
     */
    @Getter(AccessLevel.PROTECTED)
    private int maxRetryTimes = 3;

    /**
     * 消息体
     */
    @Getter
    private T message;

    /**
     * <p>消费者组</p>
     * <p>注意组名不要重复，默认组名生成规则是，类名+方法名+环境标识，一般消费者都在1个中心的话不会重复，但如果分布不同中心，存在问题。</p>
     *
     * @see AbstractRocketMqListener#generateConsumerGroup()
     */
    @Getter
    @Setter
    private String consumerGroup;

    /**
     * 是否添加tag的profile后缀
     */
    @Getter
    @Setter
    private boolean enableTagSuffix = true;

    /**
     * 是否添加consumerGroup的profile后缀
     */
    @Getter
    @Setter
    private boolean enableGroupSuffix = true;

    /**
     * 每个线程消费消息数
     * <p>注：如果consumeThreadMax和consumeThreadNumber都是1就可以单队列，不担心并发问题了</p>
     *
     * @see RocketMQMessageListener#consumeThreadNumber()
     */
    @Getter
    @Setter
    private int consumeThreadNumber;

    /**
     * 消费线程号占位符，多个线程消费时使用
     * 优先级高于 consumeThreadNumber
     * <p>例如：${rocketmq.consumer.thread-number-holder}</p>
     */
    @Getter
    @Setter
    private String consumeThreadNumberHolder;

    /**
     * 消息业务处理
     *
     * @param messageVo 业务约定的消息体
     * @return 是否成功
     */
    public abstract boolean processMessage(MessageVo messageVo);

    /**
     * 消息处理失败，重试次数超过最大重试次数
     *
     * @param message 消息体
     */
    protected void handleMaxRetryExceeded(T message) {
        // do nothing
    }

    /**
     * <p>重试消息实现</p>
     *
     * @param message 消息体
     */
    protected void retryMessage(T message) {
        // 最简重试方式，一般抛出异常就好
        throw new RuntimeException("消息处理失败，msgId: " + getMsgId());
    }

    /**
     * <p>接收消息，处理业务逻辑的主体方法</p>
     *
     * @param message 消息体
     * @return 是否成功
     */
    public boolean processMessage(T message) {

        // 入口设置消费者进程的traceId
        this.setupTraceId(message);

        this.message = message;

        try {
            if (log.isInfoEnabled()) {
                log.info("接收到MQ消息，msgId: [{}], channel: [{}], routerKey: [{}]", getMsgId(), getChannel(), getRouteKey());
            }

            final MessageVo messageVo = convertMessageVo(message);

            if (Objects.isNull(messageVo)) {
                log.error("消息体内容非约定JSON格式，数据有误。直接返回成功，不执行消费逻辑，msgId: {}", getMsgId());
                return true;
            }

            if (log.isInfoEnabled()) {
                log.info("接收到MQ消息：msgId: {}, body: {}", getMsgId(), JSONUtil.toJsonStr(messageVo));
            }
            boolean result = false;

            try {
                result = this.processMessage(messageVo);
                if (result) {
                    doOnResultSuccess(messageVo);
                } else {
                    doOnResultFailed(messageVo);
                }
            } catch (Exception e) {
                doOnMessageException(message, e);
            }
            return result;
        } finally {
            // 清理消费者进程的traceId
            clearTraceId();
        }
    }

    /**
     * <p>设置消费者进程的traceId</p>
     *
     * @param message 消息体
     */
    protected abstract void setupTraceId(@NonNull T message);

    /**
     * <p>清理消费者进程的traceId</p>
     */
    protected void clearTraceId() {
        // 清理traceId，防止影响其他线程
        String traceId = MDC.popByKey(MqChannel.TRACE_ID_KEY);
        if (Objects.nonNull(traceId)) {
            log.debug("清理消费者进程的traceId: {}", traceId);
        }
    }

    /**
     * <p>消息处理，业务失败时扩展处理操作，例如可记录日志，修改存库的消息状态等</p>
     *
     * @param messageVo 消息体
     */
    protected void doOnResultFailed(@NonNull MessageVo messageVo) {
        log.error("消息处理业务失败，msgId: {}", getMsgId());
    }

    /**
     * 消息成功后的扩展处理操作，例如可记录日志，修改存库的消息状态等
     *
     * @param messageVo 消息体
     */
    protected void doOnResultSuccess(MessageVo messageVo) {
        // 消息处理成功后的操作
    }

    /**
     * <p>是否手动确认消息</p>
     * <p>注：并不是MQ中间件都支持act</p>
     *
     * @return true，如果手动确认消息；false，如果自动确认消息
     */
    protected boolean isManualAck() {
        return true;
    }

    /**
     * 获取消息体id
     *
     * @return 消息体id
     */
    public abstract String getMsgId();

    /**
     * 主分类标识 核心业务分类（如：order, payment），对应 RocketMQ 的 Topic 或 Kafka 的主 Topic
     *
     * @return 主分类标识
     */
    public abstract String getChannel();

    /**
     * 路由分类标识 业务分类（如：order-create, order-cancel），对应 RocketMQ 的 Tag 或 Kafka 的子 Topic
     *
     * @return 路由分类标识
     */
    public abstract String getRouteKey();

    /**
     * 转换消息体为标准化消息VO
     *
     * @param message 原始消息体
     * @return 标准化消息VO
     */
    protected abstract MessageVo convertMessageVo(T message);

    public String getInstanceName() {
        return this.getClass().getSimpleName();
    }

    /**
     * <p>消息处理异常，一般是运行时接口调用异常</p>
     *
     * @param message 消息体
     * @param e       异常
     */
    protected void doOnMessageException(@NonNull T message, Exception e) {
        log.error(StrUtil.format("消息处理异常，msgId: {}", getMsgId()), e);
        // 消息处理异常，可以记录日志，也可以将消息重新放回队列
    }

    /**
     * <p>安全转换消息体中的data为List</p>
     * <p>实际 <code>messageVo.getData().toBean(List.class)</code> 也可以，但会不那么"安全" </p>
     *
     * @param messageVo 消息体
     * @param clazz     转换的对象类型
     * @param <M>       转换的对象类型
     * @return List<M>
     */
    @NonNull
    public static <M> List<M> formMessageToList(@NonNull MessageVo messageVo, Class<M> clazz) {
        JSON json = messageVo.getData();
        if (json instanceof JSONArray jsonArray) {
            if (log.isDebugEnabled()) {
                log.debug("JSON转换成List成功，json: {}", JSONUtil.toJsonStr(json));
            }
            if (jsonArray.isEmpty()) {
                return List.of();
            }
            return jsonArray.toList(clazz);
        }
        log.error("JSON转换成List失败，json: {}", json);
        throw new IllegalArgumentException("JSON转换成List失败，json: " + json);
    }

    /**
     * 转换消息体为标准化消息VO
     *
     * @param messageContent 原始消息体
     * @return 标准化消息VO
     */
    public static MessageVo convertMessageVo(String messageContent) {
        return JSONUtil.toBean(messageContent, MessageVo.class, true);
    }

    /**
     * 获取消息标签，去除环境后缀
     *
     * @return 消息标签
     */
    public String getTagWithoutSuffix() {
        throw new UnsupportedOperationException("getTagWithoutSuffix() 方法未实现，请在子类中实现该方法。");
    }
}
